#!/usr/bin/env python
# -*- coding: utf-8 -*-
# @Time    : 2023/2/19 4:28
# @Author : 闫峰
# @File    : rabbitmq_utils.py
import threading
import time
import concurrent.futures
import pika

class MyClass:
    static_var = 0

    @classmethod
    def increment_static_var(cls):
        cls.static_var += 1


class RabbitMq_:

    def __init__(self,queue_name,routing_key):
        self.lock = threading.Lock()
        with self.lock:
            self.queue_name =queue_name
            credentials = pika.PlainCredentials("admin", "Zhcx?2021")
            # 连接RabbitMQ服务器
            self.connection = pika.BlockingConnection(pika.ConnectionParameters(host='10.158.158.31',port=5672,credentials=credentials))
            self.channel = self.connection.channel()

            # 创建队列，并且开启持久化
            self.channel.queue_declare(queue=queue_name, durable=True)
            self.routing_key = routing_key
            # 初始化线程池
            self.executor = concurrent.futures.ThreadPoolExecutor(max_workers=10)  # 设置合适的线程数
    def send_message(self,message):
        # 发送消息，并且将消息设置为持久化
        with self.lock:
            self.channel.basic_publish(exchange='',
                                  routing_key=self.routing_key,
                                  body=message,
                                  properties=pika.BasicProperties(
                                      delivery_mode=2
                                  ))

    def on_message_callback(self, ch, method, properties, body,fun):
        # 将消息处理任务提交给线程池
        future = self.executor.submit(fun, body)

        # 当任务完成时，确认消息
        def on_future_done(future):
            try:
                is_processed = future.result()
                if is_processed:
                    ch.basic_ack(delivery_tag=method.delivery_tag)
                    print("Message confirmed")
                else:
                    ch.basic_nack(delivery_tag=method.delivery_tag, requeue=True)
                    print("Message processing failed, requeued")
            except Exception as e:
                # 处理异常
                print(f"Error processing message: {e}")
                ch.basic_nack(delivery_tag=method.delivery_tag, requeue=True)
                print("Message processing failed, requeued due to exception")

                # 添加回调到 future 对象，以便在任务完成时执行

        future.add_done_callback(on_future_done)

    def receive_messages(self):
        # 设置消费者允许的最大未确认消息数量为1
        self.channel.basic_qos(prefetch_count=1)

        # 定义队列的消费回调
        self.channel.basic_consume(queue=self.queue_name, on_message_callback=self.on_message_callback)

        # 开始消费消息
        print(' [*] Waiting for messages. To exit press CTRL+C')
        self.channel.start_consuming()
    def closeConnection(self):
        with self.lock:
            self.connection.close()


"""
注意：这里代码的队列名和简单模式里面的队列名是不一致的，这是因为执行简单模式下的代码之后已经创建了hello这个队列，是非持久化的，这里如果还是hello这个队列的话，那么执行的时候会报错，
因为RabbitMQ不允许使用不同的参数重新声明一个队列或交换器，需要删除重建。

注意：这里代码的routing_key也需要改成和队列名一致，否则会丢失消息。
"""